package com.seari.dao;

import com.seari.bean.DataTable;
import com.seari.common.Const;
import com.seari.service.impl.FirstInitServiceImpl;
import com.seari.utils.*;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;


public class FirstInitDaoImpl implements Runnable{

    private static Logger logger = LoggerFactory.getLogger(FirstInitDaoImpl.class);

    private static String username = PropertiesUtil.getProperty("commonConfig.properties","username");
    private static String password = PropertiesUtil.getProperty("commonConfig.properties","password");
    private static String openurl = PropertiesUtil.getProperty("commonConfig.properties","openurl");
    private static String database = PropertiesUtil.getProperty("commonConfig.properties","database");
    private static String measurement = PropertiesUtil.getProperty("commonConfig.properties","measurement");

    BlockingQueue<List<String>> aiQueue;
    BlockingQueue<List<String>> diQueue;
    Jedis jedis = JedisPoolUtils.getJedis();

    InfluxDBConnUtils influxDB = new InfluxDBConnUtils(openurl, database);
    GuavaLocalCacheUtils guavaLocalCacheUtils = new GuavaLocalCacheUtils();

    public FirstInitDaoImpl(BlockingQueue<List<String>> aiQueue, BlockingQueue<List<String>> diQueue){
        this.aiQueue = aiQueue;
        this.diQueue = diQueue;
    }

    @Override
    public void run(){
        try {
            influxDB.influxDbBuild();
            influxDB.createRetentionPolicy();
            process();
        } catch (Exception ex) {
            FirstInitServiceImpl.workerCount.decrementAndGet();
            ex.printStackTrace();
        }finally {
            jedis.disconnect();
            jedis.close();
        }
    }

    public void process() throws Exception
    {
        while (true)
        {
            List<String> diDatas = diQueue.poll(3, TimeUnit.SECONDS);
            if(null == diDatas)
            {
                List<String> aiDatas = aiQueue.poll(3, TimeUnit.SECONDS);
                if(null == aiDatas)
                {
                    logger.info("first init diQueue and aiQueue is null");
                    return;
                }else {
                    sendAiData(aiDatas);
                }
            }else {
                sendDiData(diDatas);
            }
        }
    }

    /**
     *
     * @param diDatas
     * @throws Exception
     */
    public void sendDiData(List<String> diDatas) throws Exception{

        Map<String, String> tags = new HashMap<String, String>();
        Map<String, Object> fields = new HashMap<String, Object>();
        String diDataKey = diDatas.get(diDatas.size()-1);
        diDatas.remove(diDatas.size()-1);

        Pipeline pipeline = jedis.pipelined();
        for (String diData : diDatas){
            pipeline.hget(diDataKey,diData);
        }
        List<Object> results = pipeline.syncAndReturnAll();

        BatchPoints batchPoints = BatchPoints
                .database(database)
                //.tag("async", "true") //Add a tag to this set of points.
                //  .retentionPolicy("default")
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();

        for(int i = 0 ;i<results.size();i++){
            String keyStr = diDatas.get(i);
            String valueStr = (String)results.get(i);
            String key = diDataKey+keyStr;
            List<DataTable> dataTableList =  (List<DataTable>)guavaLocalCacheUtils.cache.getIfPresent(key);
            //本地是否有 mysql缓存   DI
                if(!CommonUtils.objectIsNull(dataTableList)) {
                    //DI取反标志
                    if(Const.DI_REVERSE_YES==dataTableList.get(0).getReverse()) {
                        if("0"==valueStr){
                            valueStr="1";
                        }else{
                            valueStr="0";
                        }
                    }

                    for(DataTable dataTable:dataTableList){
                        //值匹配mysql配置  DI
                        if(valueStr.equals(dataTable.getBitAddressValueCode())) {
                            Point point = Point
                                    .measurement(measurement)
                                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                                    .tag("TAG_HASH_KEY", diDataKey)
                                    .tag("TAG_KEY", keyStr)
                                    .tag("TAG_KEY_BIT_ADDRESS", dataTable.getBitAddress())
                                    .addField("COMMMENT", dataTable.getComment())
                                    .addField("STATION_EN", dataTable.getStationEn())
                                    .addField("STATION_CN", dataTable.getStationEn())
                                    .addField("ADDRESS", dataTable.getAddress())
                                    .addField("ACTUAL_ADDRESS", dataTable.getActualAddress())
                                    .addField("RATIO", dataTable.getRatio())
                                    .addField("IS_DI", dataTable.getIsDi())
                                    .addField("SYSTEM_NAME", dataTable.getSystemName())
                                    .addField("BIT_ADDRESS_EN", dataTable.getBitAddressEn())
                                    .addField("BIT_ADDRESS_CN", dataTable.getBitAddressCn())
                                    .addField("VALUE", valueStr)
                                    .build();
                            batchPoints.point(point);
                        }
                    }
                }else{
                    logger.info("FirstInitDaoImpl.sendDiData localMysqlCache key === "+key+" === is null");
                }
            }
        List<Point> pp = batchPoints.getPoints();
        logger.info("FirstInitDaoImpl.sendDiData.pp size = " +pp.size());
        influxDB.write(batchPoints);
    }

    /**
     *
     * @param aiDatas
     * @throws Exception
     */
    public void sendAiData(List<String> aiDatas) throws Exception{

        Map<String, String> tags = new HashMap<String, String>();
        Map<String, Object> fields = new HashMap<String, Object>();
        String aiDataKey = aiDatas.get(aiDatas.size()-1);
        aiDatas.remove(aiDatas.size()-1);

        Pipeline pipeline = jedis.pipelined();
        for (String aiData : aiDatas){
            pipeline.hget(aiDataKey,aiData);
        }
        List<Object> results = pipeline.syncAndReturnAll();

        BatchPoints batchPoints = BatchPoints
                .database(database)
                //.tag("async", "true") //Add a tag to this set of points.
                //  .retentionPolicy("default")
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();

        int combineNum = 0;
        String firstValueStr = null;
        for (int i = 0; i < results.size(); i++) {
            String keyStr = aiDatas.get(i);
            String valueStr = (String) results.get(i);
            String key = aiDataKey+keyStr;
            List<DataTable> dataTableList =  (List<DataTable>)guavaLocalCacheUtils.cache.getIfPresent(key);
            aiDataHander(key,aiDataKey,keyStr,dataTableList,valueStr,batchPoints);
        }
        List<Point> pp = batchPoints.getPoints();
        logger.info("FirstInitDaoImpl.sendAiData.pp size = " +pp.size());
        influxDB.write(batchPoints);
    }

    /**
     *
     * @param key
     * @param aiDataKey
     * @param keyStr
     * @param dataTableList
     * @param valueStr
     * @param batchPoints
     */
    public void aiDataHander(String key,String aiDataKey,String keyStr,List<DataTable> dataTableList,String valueStr,BatchPoints batchPoints){
        //本地是否有mysql缓存   AI
        if(!CommonUtils.objectIsNull(dataTableList)){
            //是否合并    AI
            if(Const.REDIS_COMBINE_YES.equals(dataTableList.get(0).getCombine())) {
                logger.info("HistoryDataHandlerWorker.sendAiData combine key ");
            }else{
                //是否拆分  AI
                if(Const.NEEDSPLIT_YES.equals(dataTableList.get(0).getNeedSplit())){

                    String binaryStr =  DataUtil.getHexBinaryStr(Integer.parseInt(valueStr));

                    for(DataTable dataTable : dataTableList){
                        String value = binaryStr.substring(Integer.parseInt(dataTable.getBitAddress()), Integer.parseInt(dataTable.getBitAddress())+1);
                        for (DataTable dataTable1 : dataTableList) {
                            //拆位值匹配mysql配置  AI
                            if (dataTable.getBitAddress().equals(dataTable1.getBitAddress()) && value.equals(dataTable1.getBitAddressValueCode())) {
                                Point point = Point
                                        .measurement(measurement)
                                        .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                                        .tag("TAG_HASH_KEY", aiDataKey)
                                        .tag("TAG_KEY", keyStr)
                                        .tag("TAG_KEY_BIT_ADDRESS", dataTable.getBitAddress())
                                        .addField("COMMMENT", dataTable.getComment())
                                        .addField("STATION_EN", dataTable.getStationEn())
                                        .addField("STATION_CN", dataTable.getStationEn())
                                        .addField("ADDRESS", dataTable.getAddress())
                                        .addField("ACTUAL_ADDRESS", dataTable.getActualAddress())
                                        .addField("RATIO", dataTable.getRatio())
                                        .addField("IS_DI", dataTable.getIsDi())
                                        .addField("SYSTEM_NAME", dataTable.getSystemName())
                                        .addField("BIT_ADDRESS_EN", dataTable.getBitAddressEn())
                                        .addField("BIT_ADDRESS_CN", dataTable.getBitAddressCn())
                                        .addField("VALUE", value)
                                        .build();
                                batchPoints.point(point);
                            }
                        }
                    }
                }else{
                    for(DataTable dataTable : dataTableList) {
                        //不拆位值匹配mysql配置  AI
                        if(valueStr.equals(dataTable.getBitAddressValueCode())) {
                            Point point = Point
                                    .measurement(measurement)
                                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                                    .tag("TAG_HASH_KEY", aiDataKey)
                                    .tag("TAG_KEY", keyStr)
                                    .tag("TAG_KEY_BIT_ADDRESS", dataTable.getBitAddress())
                                    .addField("COMMMENT", dataTable.getComment())
                                    .addField("STATION_EN", dataTable.getStationEn())
                                    .addField("STATION_CN", dataTable.getStationEn())
                                    .addField("ADDRESS", dataTable.getAddress())
                                    .addField("ACTUAL_ADDRESS", dataTable.getActualAddress())
                                    .addField("RATIO", dataTable.getRatio())
                                    .addField("IS_DI", dataTable.getIsDi())
                                    .addField("SYSTEM_NAME", dataTable.getSystemName())
                                    .addField("BIT_ADDRESS_EN", dataTable.getBitAddressEn())
                                    .addField("BIT_ADDRESS_CN", dataTable.getBitAddressCn())
                                    .addField("VALUE", valueStr)
                                    .build();
                            batchPoints.point(point);
                        }
                    }
                }
            }
        }else{
            logger.info("FirstInitDaoImpl.aiDataHander localMysqlCache key === "+key+" === is null");
        }
    }
}
